package com.example.kafka.service;

import com.example.kafka.mock.MockData;
import com.example.kafka.util.KafkaRetryUtils;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.errors.RetriableException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.concurrent.*;

/**
 * Kafka同步消息发送Service
 */
@Service
public class KafkaSyncService {

    @Autowired
    private Producer<String, String> producer;

    /**
     * 用例1：标准同步发送，成功与异常分流
     */
    public String sendSyncStandard() throws Exception {
        ProducerRecord<String, String> record = new ProducerRecord<>(MockData.TOPIC, MockData.KEY, MockData.VALUE);
        try {
            RecordMetadata metadata = producer.send(record).get(3, TimeUnit.SECONDS);
            return "发送成功: " + metadata.offset();
        } catch (ExecutionException e) {
            if (e.getCause() instanceof RetriableException) {
                // 可重试异常
                return "可重试异常: " + e.getMessage();
            } else {
                // 不可重试异常
                throw e;
            }
        } catch (TimeoutException e) {
            // 超时异常
            return "超时异常: " + e.getMessage();
        }
    }

    /**
     * 用例2：同步发送+自定义重试
     */
    public String sendSyncWithRetry() throws Exception {
        ProducerRecord<String, String> record = new ProducerRecord<>(MockData.TOPIC, MockData.KEY, MockData.VALUE);
        return KafkaRetryUtils.retry(3, 200, () -> {
            try {
                RecordMetadata metadata = producer.send(record).get(2, TimeUnit.SECONDS);
                return "发送成功: " + metadata.offset();
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        });
    }

    /**
     * 用例3：同步发送+超时降级
     */
    public String sendSyncWithTimeoutFallback() {
        ProducerRecord<String, String> record = new ProducerRecord<>(MockData.TOPIC, MockData.KEY, MockData.VALUE);
        try {
            producer.send(record).get(1, TimeUnit.SECONDS);
            return "发送成功";
        } catch (TimeoutException e) {
            // 超时降级处理
            return "降级：消息发送超时";
        } catch (Exception e) {
            return "异常：" + e.getMessage();
        }
    }

    /**
     * 用例4：同步发送+不可重试异常补偿
     */
    public String sendSyncWithCompensation() {
        ProducerRecord<String, String> record = new ProducerRecord<>(MockData.TOPIC, MockData.KEY, MockData.VALUE);
        try {
            producer.send(record).get();
            return "发送成功";
        } catch (ExecutionException e) {
            if (!(e.getCause() instanceof RetriableException)) {
                // 记录到补偿队列
                compensate(record, e.getMessage());
                return "已补偿";
            }
            return "可重试异常";
        } catch (Exception e) {
            return "其他异常";
        }
    }

    /**
     * 用例5：同步发送+异常类型细分处理
     */
    public String sendSyncWithErrorType() {
        ProducerRecord<String, String> record = new ProducerRecord<>(MockData.TOPIC, MockData.KEY, MockData.VALUE);
        try {
            producer.send(record).get();
            return "发送成功";
        } catch (ExecutionException e) {
            Throwable cause = e.getCause();
            if (cause instanceof RetriableException) {
                return "可重试异常";
            } else if (cause instanceof org.apache.kafka.common.errors.AuthorizationException) {
                return "鉴权异常";
            } else {
                return "其他不可重试异常";
            }
        } catch (Exception e) {
            return "未知异常";
        }
    }

    /**
     * 补偿逻辑（示例）
     */
    private void compensate(ProducerRecord<String, String> record, String reason) {
        // TODO: 实现补偿逻辑，如写入本地消息表
    }
}
